/*
 * CStreamLayer.cpp
 *
 *  Created on: 2011-9-20
 *      Author: stargui
 */

#include "CStreamLayer.h"
#include "CSessionManager.h"

CStreamLayer::CStreamLayer()
{
	taskHandle						= -1;
	taskindex						= 0;
	source							= 0x80000000;
	passiveMode					= true;
	_threadid						= NULL;
	status							= -1;
	m_remoteplayback				= 0;
	m_remoteplaybackchcount		= 0;
	taskName						= "";
	sessionid						= "";
	username						= "";
	password						= "";

	memset(&m_pPack,0,sizeof(SShareMemData));
	m_pPack.pstuPack[0].pAddr	= NULL;
	m_pPack.pstuPack[1].pAddr	= NULL;

	_linktype						= 0;
	m_ipport						= "";


	myList.clear();
	_operation.clear();

	_operation[PAULOAD_REMOTE_PLAYBACK_TYPE_EM] 	= boost::bind(&CStreamLayer::CtrlRemotePlayback,this,_1);

}

CStreamLayer::~CStreamLayer()
{
	std::map<unsigned int, taskSession_node *> ::iterator it;
	_taskmutex.ThreadLock();
	status										= 1;
	for (it = myList.begin(); it != myList.end(); it++)
	{
		if(myList[it->first])
		{
			myList[it->first]->stopflag		= 1;
		}
	}
	_taskmutex.ThreadUnlock();
	while(1)
	{
		_taskmutex.ThreadLock();
		if(myList.empty())
		{
			_taskmutex.ThreadUnlock();
			CPSelectUtils::mSleep(50);
			break;
		}
		_taskmutex.ThreadUnlock();
		CPSelectUtils::mSleep(20);
	}

	DEBUG_WARNING("--media--thread--exit!!-pid:[%d]-\n",getpid());
	this->InternalStop();

}

std::map<unsigned int, taskSession_node*> CStreamLayer::GetTaskList()
{
	std::map<unsigned int, taskSession_node*>		list;
	_taskmutex.ThreadLock();
	list					= myList;
	_taskmutex.ThreadUnlock();

	return list;
}



int CStreamLayer::TimeStringToDateInfo(std::string timestring,DateInfo_Tag &datainfo)
{
	if(timestring.size() != 14)
	{
		return -1;
	}
	datainfo.date_YMD.year	= atoi(timestring.substr(0,4).c_str()) % 1000;
	datainfo.date_YMD.month 	= atoi(timestring.substr(4,2).c_str());
	datainfo.date_YMD.day	= atoi(timestring.substr(6,2).c_str());
	datainfo.date_HMS.hour	= atoi(timestring.substr(8,2).c_str());
	datainfo.date_HMS.minute	= atoi(timestring.substr(10,2).c_str());
	datainfo.date_HMS.second	= atoi(timestring.substr(12,2).c_str());
	printf("------[%d]--[%d]--[%d]----\n",datainfo.date_YMD.year,datainfo.date_YMD.month,datainfo.date_YMD.day);
	printf("------[%d]--[%d]--[%d]----\n",datainfo.date_HMS.hour,datainfo.date_HMS.minute,datainfo.date_HMS.second);
	return 0;
}

//�ش����������
void CStreamLayer::SetTaskSessionCallback(boost::function<bool(unsigned int,unsigned int, Json::Value &,int)> callback)
{
	//_sessionCallback = callback;
}
taskSession_node* CStreamLayer::QuerySession(unsigned int task_ssid)
{
	taskSession_node	*currentsession	= NULL;
	std::map<unsigned int, taskSession_node *> ::iterator it;

	it										= myList.find(task_ssid);
	if(it != myList.end())
	{
		currentsession					= it->second;
	}
	return currentsession;
}


void *CStreamLayer::ThreadFuncImpl()
{
	if(myList.empty())
	{
		if(1 != status)
		{
			CPSelectUtils selectutils;
			selectutils.InsertfdSets(taskHandle, P_FD_SETS_READ);
			selectutils.InsertfdSets(taskHandle, P_FD_SETS_ERROR);
			int		iresult	= selectutils.Select(10000);
			if (iresult  > 0)
			{
				if(selectutils.CheckSets(taskHandle, P_FD_SETS_READ))
				{
					char _buf[12]				= {0};
					if(CPSocketUtils::Recv(taskHandle,_buf,12,0) <= 0)
					{
						status		= 1;
					}
				}
				else if (selectutils.CheckSets(taskHandle, P_FD_SETS_ERROR))
				{
					status		= 1;
				}
			}
			else if(iresult < 0)
			{
				status		= 1;
			}
			CPSelectUtils::mSleep(10);
		}
		else
		{
			CPSelectUtils::mSleep(10);
		}
		return NULL;
	}

	taskSession_node	*currentsession	= NULL;
	std::map<unsigned int, taskSession_node *> ::iterator it;
	std::map<unsigned int, taskSession_node *> ::iterator itnext;
	std::map<PAYLOAD_TYPE, boost::function<int(taskSession_node*)> >::iterator iter;

	for (it = myList.begin(),itnext = it; it != myList.end();)
	{
		_taskmutex.ThreadLock();
		if(!it->second)
		{
			itnext		= it++;
			_taskmutex.ThreadUnlock();
			continue;
		}

		if(status	== 1)
		{
			itnext						= it;
			itnext++;
			StopNetServerAndSend(it->second);
			CSessionManager::GetInstance()->FreeTaskNode(it->second->nodeindex);
			it->second					= NULL;
			myList.erase(it);
			it							= itnext;
			_taskmutex.ThreadUnlock();
			//CPSelectUtils::mSleep(30);
			continue;
		}

		currentsession			  	= it->second;
		if(currentsession->stopflag)
		{
			itnext						= it;
			itnext++;
			StopNetServerAndSend(currentsession);
			if(2 == currentsession->stopflag)
			{
				status		= 1;
			}
			CSessionManager::GetInstance()->FreeTaskNode(currentsession->nodeindex);
			currentsession			= NULL;
			it->second					= NULL;
			myList.erase(it);
			it							= itnext;
			_taskmutex.ThreadUnlock();

			continue;
		}
		 it++;

		_taskmutex.ThreadUnlock();
		if((iter = _operation.find((PAYLOAD_TYPE)currentsession->pt)) != _operation.end())
		{
			if (iter->second( currentsession))
			{
				currentsession->stopflag	= 1;
			}
			else
			{
				if(currentsession->stopflag	== 2)
				{
					_taskmutex.ThreadLock();
					status		= 1;
					it 			= myList.begin();
					_taskmutex.ThreadUnlock();
				}
			}
		}
	}
	return NULL;
}
int CStreamLayer::StopNetServerAndSend(taskSession_node *currentsession)
{
	Json::Value response;
	int result 									= -1;
	switch((PAYLOAD_TYPE)(currentsession->pt))
	{
		case PAYLOAD_SIGNAL:
			break;
		case PAYLOAD_METADATA:
			break;
		case PAYLOAD_VIDEO_TYPE_EM:
			break;
		case PAULOAD_REMOTE_PLAYBACK_TYPE_EM:
			{
				StopRemotePlayBack(currentsession->ssid,currentsession);
				m_remoteplaybackchcount		= 0;
				m_remoteplayback				= 0;
			}
			break;
		default:
			break;
	}
	return result;
}
/*****************************************************************************************************************************************************/
/**
 * send all date to client
 * bug
 */
bool write_to_client(int filedes, const void *buffer, size_t size)
{

	size_t w_counter=0,write_amout=size,w_num=0;

	const void* start_buffer=buffer;
	while(true)
	{
		w_num=write(filedes,buffer,size);
		if(w_num==-1)
		{
			if(EBADF==errno)  //if the socket is colse
			{
				return false;
			}else   //for common error
			{
				CHandleException::print_socket_excepetion("write ");
			}
			continue;
		}
		w_counter+=w_num;
		if(w_counter==write_amout)
		{
			//printf(".");
			break;
		}else
		{
			buffer=start_buffer+w_counter;
			size=write_amout-w_counter;
		}
	}
	return true;
}
/*
 * put in thread pool
 * */
void* ThreadFun(void*param)
{

	std::cout<<"jmp in ThreadFun \n ";
	int i=0;
	THREADPARA* threadpara=(THREADPARA*)param;
//	char* msgarry=new char[(MegLen-1)*threadpara->_channel_and_time.channel];
	char msgarry[MegLen]; //the first byte is cmd id ,the rest is extend data
	if(msgarry==NULL)
	{
		printf("error at get msgarry \n");
		return (void*)NULL;
	}
	int channnel_num=4;
	char *channel_record=new char[channnel_num];  //for every channel state
	memset(channel_record,1,channnel_num);  //1 is open
	memset((void*)msgarry,1,(MegLen));
	int threadflag=channnel_num;				//  number of channels
	GetFrameFromFile *from_file=new GetFrameFromFile[channnel_num];  // now only one
	FramFlag flag[3];
	flag[0].type='i';
	flag[0].flag[0]='2';
	flag[0].flag[1]='d';
	flag[0].flag[2]='c';
	flag[1].type='p';
	flag[1].flag[0]='3';
	flag[1].flag[1]='d';
	flag[1].flag[2]='c';
	flag[2].type='a';
	flag[2].flag[0]='4';
	flag[2].flag[1]='d';
	flag[2].flag[2]='c';
	N9_NET_HEAD net_head;
	net_head.nReserve='R';
	net_head.byType=4;
	from_file[0].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
	from_file[1].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
	from_file[2].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
	from_file[3].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
//	from_file[4].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
//	from_file[5].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
//	from_file[6].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
//	from_file[7].inite("/home/bird/Desktop/record2.264",(STime*)NULL,flag,4); //file path
	while(threadflag)
	{
		long int _message;char _cmdid;
		for(i=0;i<10;i++)  // 10 最大读取10
		{

			if(AdpMsg::MsgRcv(threadpara->socketfd,_cmdid,_message)==-1) //read all
			{
				break;
			}else
			{
				msgarry[0]=_cmdid; //channel cmdid
				if(_cmdid==3||_cmdid==6)
				{
					memcpy((void*)(&msgarry[1]),(void*)(_message),MegLen-1);
				}
			}
		}
		char type;int len;
		for(i=0;i<channnel_num;i++) // 循环检测每个通道的消息消息矩阵并执行相应操作
		{
			if(channel_record[i]==0)
			{
				break;
			}
			switch(msgarry[0]) // i*(MegLen-1)*1 is channel open or close | i*(MegLen-1)*1+1 cmd
			{
				case 0: //stop
					threadflag--;
					msgarry[i*MegLen]=0;// close this channel
					break;
				case 1: //continue
				{   //send datapacket
					net_head.ssrc=(short)i;
					if(NULL==from_file[i].get_frame(&type,&len))
					{
						CHandleException::print("meet eof !");
						channel_record[i]=0;
						threadflag--;
						printf("active close channel %d\n ",i);
						break;
					}
					net_head.nLen=htonl(len);
					write_to_client(threadpara->socketfd,(void*)&net_head,sizeof(N9_NET_HEAD));
					write_to_client(threadpara->socketfd,(void*)from_file[i].data_buffer,len);
				}
					break;
				case 2://pause
//					printf(" server execute pause command channel %d\n ",i);
					break;
				case 3://drag
					break;
				case 4: // lost p frame
				{
					net_head.ssrc=(short)i;
					while(true) //only send  i pframe
					{
						if(from_file[i].get_frame(&type,&len)==NULL)
						{
							CHandleException::print(" meet EOF \n");
							msgarry[i*MegLen]=0;// close this channel
							break;
						}
						if(type=='i')
						{
							break;
						}
					}
					net_head.nLen=htonl(len);
					write_to_client(threadpara->socketfd,(void*)&net_head,sizeof(N9_NET_HEAD));
					write_to_client(threadpara->socketfd,(void*)from_file[i].data_buffer,len);
				}
					break;
				case 5: //normal
					break;
				case 6: //倍数播放
					break;
				default:
					break;
			}
		}
	}
	printf("thread leaving \n ");
	close(threadpara->socketfd); //主动关闭该套接字
	delete threadpara;
	return (void*)NULL;

}

int CStreamLayer::datalink_interface(const int i_taskhandle,const channel_and_time remotequest)
{
	THREADPARA* threadpara=new THREADPARA();
	threadpara->socketfd=i_taskhandle;
	threadpara->_channel_and_time=remotequest;
	ThreadPool::T_AddJop(ThreadFun,(void*)threadpara);
	return 0;
}

int CStreamLayer::msglink_interface(const int i_taskhandle,int i_cmd,long int i_optional_para)
{
	assert( AdpMsg::MsgSend(i_cmd,i_taskhandle,i_optional_para)==0);
	return 0;
}
/*****************************************************************************************************************************************************/
void CStreamLayer::RequestRemotePlayback(Json::Value  request)
{//根据协议解析各个字段，对必须的字段需要校验否则解析协议失败
	int		paramiserror					= 0;
	Json::Value response;
	response[SESSION] 					= request[SESSION];
	sessionid								= request[SESSION].asString();
	response[MODULE] 						= MODULE_MEDIASTREAM;
	response[OPERATION] 					= request[OPERATION];
	response[RESPONSE][MEDIASTREAM_STREAMNAME]	= request[PARAMETER][MEDIASTREAM_STREAMNAME];

	if(!m_remoteplayback)
	{
		m_remoteplayback					= 1;
		m_remoteplaybackchcount			= 0;
	}
	else
	{
		response[RESPONSE][ERRORCODE]	= TASKEXIST;
		response[RESPONSE][ERRORCAUSE]	= EP_CN_TASKEXIST;
		//_sessionCallback(source,taskindex,response,0);
		return;
	}

	if(request[PARAMETER][MEDIASTREAM_PT].isNull())
	{
		paramiserror						= 1;
	}
	else
   {
		if(request[PARAMETER][MEDIASTREAM_PT].asInt() != PAULOAD_REMOTE_PLAYBACK_TYPE_EM )
		{
			paramiserror					= 1;
		}
	}

	DateInfo_Tag	 starttime,endtime;
	memset(&starttime,0,sizeof(DateInfo_Tag));
	memset(&endtime,0,sizeof(DateInfo_Tag));

	std::string   starttimestring = "",endtimestring = "";
	if(request[PARAMETER][MEDIASTREAM_STARTTIME].isNull())
	{
		response[RESPONSE][ERRORCODE]  = TIME_ERROR;
		response[RESPONSE][ERRORCAUSE] = EP_CN_TIME_ERROR;
		//_sessionCallback(source,taskindex,response,0);
		return ;
	}
	starttimestring						= request[PARAMETER][MEDIASTREAM_STARTTIME].asString();

	if(!request[PARAMETER][MEDIASTREAM_ENDTIME].isNull())
	{
		endtimestring						= request[PARAMETER][MEDIASTREAM_ENDTIME].asString();
	}

	if(TimeStringToDateInfo(starttimestring,starttime))
	{
		response[RESPONSE][ERRORCODE]  = TIME_ERROR;
		response[RESPONSE][ERRORCAUSE] = EP_CN_TIME_ERROR;
		//_sessionCallback(source,taskindex,response,0);
		return ;
	}

	if(!endtimestring.empty())
	{
		if(TimeStringToDateInfo(endtimestring,endtime))
		{
			response[RESPONSE][ERRORCODE]	= TIME_ERROR;
			response[RESPONSE][ERRORCAUSE]	= EP_CN_TIME_ERROR;
			//_sessionCallback(source,taskindex,response,0);
			return ;
		}
	}

	unsigned int	channel	= 0;
	unsigned int bakchannel	= 0;
	if(!request[PARAMETER][MEDIASTREAM_CHANNEL].isNull())
	{
		channel				= request[PARAMETER][MEDIASTREAM_CHANNEL].asInt();

		if(HARDWARE_VIDEO_CHANNEL_NUM == 1)
		{
			channel &= 0x01;
		}
		else if(HARDWARE_VIDEO_CHANNEL_NUM == 4)
		{
			channel &= 0x0F;
		}
		else if(HARDWARE_VIDEO_CHANNEL_NUM == 8)
		{
			channel &= 0xFF;
		}
		else if(HARDWARE_VIDEO_CHANNEL_NUM == 16)
		{
			channel &= 0xFFFF;
		}

		bakchannel								= channel;
		unsigned int			channelmax		= 0;
		if(passiveMode)
		{//权限校验，因为有的用户只对部分通道由回放权限
/*			if(CSessionManager::GetInstance()->CheckRightFun(username,password,FunRightType_RemotePlayback,(void *)&channel,channelmax))
			{
				response[RESPONSE][ERRORCODE]  	= NORIGHT;
				response[RESPONSE][ERRORCODE]  	= NORIGHT;
				response[RESPONSE][ERRORCAUSE] 	= EP_CN_NORIGHT;
				//_sessionCallback(source,taskindex,response,0);
				DEBUG_WARNING("--------------no----right--------------------\n");
				return;
			}*/

			DEBUG_WARNING("-----channelmax:[%02x]-------\n",channelmax);
			channel				= channel & channelmax;
			if(bakchannel & (~channelmax ))
			{
				/*response[RESPONSE][ERRORCODE]  	= NORIGHT;*/
				response[RESPONSE][ERRORCODE]  	= SUCCESS_OK;
			/*	response[RESPONSE][ERRORCAUSE] 	= EP_CN_NORIGHT;*/
				response[RESPONSE][ERRORCAUSE] 	= EP_CN_SUCCESS_OK;
				////_sessionCallback(source,taskindex,response,0);
				DEBUG_WARNING("----------some--channel--no----right--------------------\n");
			}

		}//统计用户一次回放了多少个通道

		for(int index = 0;index < HARDWARE_VIDEO_CHANNEL_NUM;index++)
		{
			if(channel & (0x01 << index))
			{
				m_remoteplaybackchcount++;

				//DEBUG_WARNING("---------m_remoteplaybackchcount:[%d]------------\n",m_remoteplaybackchcount);
			}
		}



	}
	int			questtype		= 0;
	if(!request[PARAMETER][MEDIASTREAM_STREAMTYPE].isNull())
	{
		questtype				= request[PARAMETER][MEDIASTREAM_STREAMTYPE].asInt();
	}

	channel_and_time		remotequest;
	memset(&remotequest,0,sizeof(channel_and_time));
	remotequest.channel		= channel;
	datetime_t		tmptime;
	memset(&tmptime,0,sizeof(datetime_t));
	tmptime.year				= starttime.date_YMD.year;
	tmptime.month				= starttime.date_YMD.month;
	tmptime.day				= starttime.date_YMD.day;
	tmptime.hour				= starttime.date_HMS.hour;
	tmptime.minute			= starttime.date_HMS.minute;
	tmptime.second			= starttime.date_HMS.second;
	//CSystemTime::DateTimeToUtcTime
	remotequest.starttime=tmptime;
	tmptime.year				= endtime.date_YMD.year;
	tmptime.month				= endtime.date_YMD.month;
	tmptime.day				= endtime.date_YMD.day;
	tmptime.hour				= endtime.date_HMS.hour;
	tmptime.minute			= endtime.date_HMS.minute;
	tmptime.second			= endtime.date_HMS.second;
	//CSystemTime::DateTimeToUtcTime(&tmptime,&remotequest.endtime);
	remotequest.endtime=tmptime;
	remotequest.streamtype	= questtype;
	datalink_interface(taskHandle,remotequest);
}
//处理回放任务
int CStreamLayer::CtrlRemotePlayback(taskSession_node *currentsession)
{
	if(0 == currentsession->stopflag)
	{//底层是否已经根据申请参数准备好回放
		if(!currentsession->isready)
		{
			CPSelectUtils::mSleep(50);
			return 0;
		}

		if(currentsession->newflag)
		{
			CPSelectUtils::mSleep(100);
			_taskmutex.ThreadLock();
			currentsession->newflag									= 0;
			_taskmutex.ThreadUnlock();
			currentsession->getTimeBak								= 0;


			if(0 == currentsession->restflag)
			{
				Json::Value response;
				response[SESSION] 									= sessionid;
				response[MODULE] 										= MODULE_MEDIASTREAM;
				response[OPERATION] 									= MEDIATASKSTART;
				response[PARAMETER][MEDIASTREAM_PT]				= currentsession->pt;
				response[PARAMETER][MEDIASTREAM_SSRC]				= currentsession->ssrc;
				response[PARAMETER][MEDIASTREAM_STREAMNAME] 		= taskName;

				char tempip[64]										= {0};
				snprintf(tempip,64,"%s:%d",(char*)currentsession->ip,currentsession->port);
				std::string ip										= "";
				ip.append(tempip);
				response[PARAMETER][MEDIAIPANDPORT]				= ip;

		//		//_sessionCallback(source,taskindex,response,0);
			}

			currentsession->restflag								= 0;
			Json::Value Response;
			Response[MODULE] 										= MODULE_MEDIASTREAM;
			Response[OPERATION] 									= MEDIA_REMOTEPLAYBACKSTART;
			Response[SESSION] 									= sessionid;
			Response[RESPONSE][MEDIASTREAM_STREAMNAME]		= taskName;
			Response[RESPONSE][MEDIASTREAM_CSRC]				= "";
			Response[RESPONSE][MEDIASTREAM_PT] 				= currentsession->pt;
			Response[RESPONSE][ERRORCODE]  						= SUCCESS_OK;
			Response[RESPONSE][ERRORCAUSE] 						= EP_CN_SUCCESS_OK;
			Response[RESPONSE][MEDIA_REST]						= 1;
			Response[RESPONSE][MEDIASTREAM_STREAMTYPE]		= currentsession->streamtype;
			CProtocolFactory packet;
			std::string sender 									= "";
			sender = packet.Serialize(0, currentsession->ssrc,Response);
			if(CPSocketUtils::Send(taskHandle,sender.c_str(),sender.length(),0,5000000,0) < 0)
			{
				_taskmutex.ThreadLock();
				currentsession->stopflag	= 1;
				_taskmutex.ThreadUnlock();
				return -1;
			}

		}
		switch(currentsession->cmdflag)
		{
			case 0://ֹͣ停止
				{
					DEBUG_WARNING("------------------stop------------------------\n");
					return 0;
				}
				break;
			case 2://没有数据，底层正在读取数据
				{
					unsigned int		channelstate = 0;
					//DEBUG_WARNING("-----------------wait-------------------------\n");
					//检测各个通道的读取数据状态，并且将通道的数据改变状态实时通知给上层
					if(!GetRemotePlayBackChnoState(currentsession,channelstate,currentsession->channel))
					{
						if(currentsession->getTimeBak++ > 100)
						{
							CProtocolFactory packet;
							std::string sender ;
							//ʵʱ���͸���ͨ�������״̬

							if(currentsession->channelmask != channelstate)
							{
							//	DEBUG_WARNING("----------channelmask:[%02X]---------------------channelstate:[%02X]-------\n",currentsession->channelmask,channelstate);
								currentsession->channelmask 				= channelstate;
								Json::Value response;
								response[SESSION] 							= sessionid;
								response[MODULE] 								= MODULE_MEDIASTREAM;
								response[OPERATION] 							= MEDIA_REMOTECHANNELSTATUS;
								response[PARAMETER][MEDIASTREAM_PT]		= currentsession->pt;
								response[PARAMETER][MEDIASTREAM_SSRC] 		= currentsession->ssrc;
								response[PARAMETER][MEDIASTREAM_CHANNEL]	= currentsession->channel;
								response[PARAMETER][MEDIACHANNELMASK]		= channelstate;
								sender											=  packet.Serialize(0, currentsession->ssrc,response );
								CPSocketUtils::Send(taskHandle,sender.c_str(),sender.length(),0,500000,0);
							}
						}
					}
					else
					{
						DEBUG_WARNING("-----------------------------------\n");
						return -1;
					}

					CPSelectUtils::mSleep(20);
					return 0;
				}
				break;
			case 3://拖拽
				{
					DEBUG_WARNING("----------------seek--------------------------\n");
					if(SeekRemotePlayBack(currentsession,currentsession->startremotetime))
					{
						currentsession->stopflag							= 1;
						Json::Value response;
						response[SESSION] 								= sessionid;
						response[MODULE] 									= MODULE_MEDIASTREAM;
						response[OPERATION] 								= MEDIAREGISTEFAILACK;
						response[PARAMETER][MEDIASTREAM_PT]			= currentsession->pt;
						response[PARAMETER][MEDIASTREAM_SSRC]			= currentsession->ssrc;
						response[PARAMETER][ERRORCODE]					= NN_SEEK_FAILED;
						response[PARAMETER][ERRORCAUSE]					= EP_CN_NN_SEEK_FAILED;
						response[PARAMETER][MEDIASTREAM_STREAMNAME] 	= taskName;

						char tempip[64]									= {0};
						snprintf(tempip,64,"%s:%d",(char*)currentsession->ip,currentsession->port);
						std::string ip									= "";
						ip.append(tempip);
						response[PARAMETER][MEDIAIPANDPORT]			= ip;
					//	//_sessionCallback(source,taskindex,response,0);
						return 0;
					}
					_taskmutex.ThreadLock();
					currentsession->cmdflag								= 1;
					currentsession->ifirst								= 0;
					_taskmutex.ThreadUnlock();
					currentsession->restflag								= 1;
					currentsession->intervaltime						= 0;
				}
				break;
		}


		if(0	== currentsession->restflag)
		{//获取数据
			int result  = GetRemotePalybackDataAndSend(currentsession);

			switch(result)
			{
				case -1://��ȡ���ʧ��
					{
						currentsession->stopflag							= 1;
						Json::Value response;
						response[SESSION] 								= sessionid;
						response[MODULE] 									= MODULE_MEDIASTREAM;
						response[OPERATION] 								= MEDIATASKSTOP;
						response[PARAMETER][MEDIASTREAM_PT]	 		= currentsession->pt;
						response[PARAMETER][MEDIASTREAM_SSRC]			= currentsession->ssrc;
						response[PARAMETER][ERRORCODE]  				= NN_GET_DATA_FAILED;
						response[PARAMETER][ERRORCAUSE] 	 			= EP_CN_NN_GET_DATA_FAILED;
						response[PARAMETER][MEDIASTREAM_STREAMNAME] 	= taskName;

						char tempip[64]									= {0};
						snprintf(tempip,64,"%s:%d",(char*)currentsession->ip,currentsession->port);
						std::string ip									= "";
						ip.append(tempip);
						response[PARAMETER][MEDIAIPANDPORT]			= ip;
					//	//_sessionCallback(source,taskindex,response,0);
					}
					break;
				case -2:////�����
					{
						currentsession->stopflag							= 1;
						Json::Value response;
						response[SESSION] 								= sessionid;
						response[MODULE] 									= MODULE_MEDIASTREAM;
						response[OPERATION] 								= MEDIATASKSTOP;
						response[PARAMETER][MEDIASTREAM_PT]	 		= currentsession->pt;
						response[PARAMETER][MEDIASTREAM_SSRC]	 		= currentsession->ssrc;
						response[PARAMETER][ERRORCODE]  				= SUCCESS_OK;
						response[PARAMETER][ERRORCAUSE]  				= EP_CN_SUCCESS_OK;
						response[PARAMETER][MEDIASTREAM_STREAMNAME] 	= taskName;

						char tempip[64]									= {0};
						snprintf(tempip,64,"%s:%d",(char*)currentsession->ip,currentsession->port);
						std::string ip									= "";
						ip.append(tempip);
						response[PARAMETER][MEDIAIPANDPORT]			= ip;
			//			//_sessionCallback(source,taskindex,response,0);

					}
					break;
				case -3://ý����·�쳣
					{
						currentsession->stopflag							= 2;
						Json::Value response;
						response[SESSION] 								= sessionid;
						response[MODULE] 									= MODULE_MEDIASTREAM;
						response[OPERATION] 								= MEDIATASKSTOP;
						response[PARAMETER][MEDIASTREAM_PT]	 		= currentsession->pt;
						response[PARAMETER][MEDIASTREAM_SSRC]	 		= currentsession->ssrc;
						response[PARAMETER][ERRORCODE] 		 			= SENDDATAFAILED;
						response[PARAMETER][ERRORCAUSE]  				= EP_CN_SENDDATAFAILED;
						response[PARAMETER][MEDIASTREAM_STREAMNAME] 	= taskName;

						char tempip[64]									= {0};
						snprintf(tempip,64,"%s:%d",(char*)currentsession->ip,currentsession->port);
						std::string ip									= "";
						ip.append(tempip);
						response[PARAMETER][MEDIAIPANDPORT]			= ip;
					//	//_sessionCallback(source,taskindex,response,0);

					}
					break;
			}
		}
		else if(2 == currentsession->restflag )
		//��ȫ��p֡
		{
			ForwardRemotePlayBack(currentsession,2);
			currentsession->restflag								= 0;
		}
		else if(3 == currentsession->restflag)
		//�ָ���
		{
			ForwardRemotePlayBack(currentsession,3);
			currentsession->restflag								= 0;
		}
		else
		{
			CPSelectUtils::mSleep(100);
			//������ֹͣ
			if(!CheckRemotePlayBackResult(currentsession))
			{
				currentsession->stopflag							= 1;
				Json::Value response;
				response[SESSION] 								= sessionid;
				response[MODULE] 									= MODULE_MEDIASTREAM;
				response[OPERATION] 								= MEDIATASKSTOP;
				response[PARAMETER][MEDIASTREAM_PT]	 		= currentsession->pt;
				response[PARAMETER][MEDIASTREAM_SSRC]	 		= currentsession->ssrc;
				response[PARAMETER][ERRORCODE]  				= SUCCESS_OK;
				response[PARAMETER][ERRORCAUSE]  				= EP_CN_SUCCESS_OK;
				response[PARAMETER][MEDIASTREAM_STREAMNAME] 	= taskName;

				char tempip[64]									= {0};
				snprintf(tempip,64,"%s:%d",(char*)currentsession->ip,currentsession->port);
				std::string ip									= "";
				ip.append(tempip);
				response[PARAMETER][MEDIAIPANDPORT]			= ip;
			//	//_sessionCallback(source,taskindex,response,0);
				return -2;//�����
			}
		}
	}
	return 0;
}



int CStreamLayer::StartRemotePlaybackAndSend(unsigned int ssid,unsigned int channel,
								StreamType streamtype,VideoType videotype,DateInfo_Tag starttime,
								DateInfo_Tag endtime,taskSession_node*	currentsession)
{
	int	result											= -1;
	Json::Value response;
	response[SESSION] 								= sessionid;
	response[MODULE] 									= MODULE_MEDIASTREAM;
	response[OPERATION] 								= MEDIAREMOTEPLAYBACK_REQUEST;
	response[RESPONSE][MEDIASTREAM_STREAMNAME]	= taskName;
	if(!channel)
	{
		response[RESPONSE][ERRORCODE]				= CHANNELILLEGAL;
		response[RESPONSE][ERRORCAUSE]				= EP_CN_CHANNELILLEGAL;
		//_sessionCallback(source,taskindex,response,0);
		return	result;
	}
//调用底层的封装接口
	result = StartRemotePlayBack(ssid,channel,streamtype,videotype,starttime,endtime,currentsession);

	if(result)
	{
		switch(result)
		{
			case -1:
				{
					response[RESPONSE][ERRORCODE]	= NN_START_PLAYBACK_FAILED;
					response[RESPONSE][ERRORCAUSE]	= EP_CN_NN_START_PLAYBACK_FAILED;
				}
				break;
			case -2:
				{
					response[RESPONSE][ERRORCODE]	= LACKRESOURCEORTASKFULL;
					response[RESPONSE][ERRORCAUSE]	= EP_CN_LACKRESOURCEORTASKFULL;
				}
				break;
			case -3:
				{
					response[RESPONSE][ERRORCODE]	= TASKEXIST;
					response[RESPONSE][ERRORCAUSE]	= EP_CN_TASKEXIST;
				}
				break;
			default:
				{
					response[RESPONSE][ERRORCODE]	= UNKNOW;
					response[RESPONSE][ERRORCAUSE]	= EP_CN_UNKNOW;
				}
				break;
		}
		//_sessionCallback(source,taskindex,response,0);
		return	result;
	}

	return result;
}

int CStreamLayer::GetRemotePalybackDataAndSend(taskSession_node *currentsession)
{
	unsigned char *databuffirst 	= NULL,
		*databufsecond 				= NULL;
	int		datalenfirst				= 0,
			datalensecond 			= 0;
	int		formattype					= 0;
	int 	channel					= 0;
	CProtocolFactory packet;
	std::string sender ;
	unsigned int		channelstate = 0;
	//检测数据是否全部读取完毕
	if(!CheckRemotePlayBackResult(currentsession))
	{
		DEBUG_WARNING("--------------------check--data--is--over---------------\n");
		return -2;//�����
	}

	//time_t		currenttime	= time(NULL);
	//DEBUG_WARNING("-------%d------------\n",abs((int)difftime(currentsession->intervaltime,currenttime)));
	//if(abs((int)difftime(currentsession->intervaltime,currenttime)) > 2)
	if(currentsession->getTimeBak++ > 40 || !currentsession->ifirst)//about 2s
	{
		currentsession->getTimeBak = 0;
	//	DEBUG_WARNING("-------%d------------\n",abs((int)difftime(currentsession->intervaltime,currenttime)));
		//currentsession->intervaltime		= currenttime;
		//检测各个通道的数据状态
		if(!GetRemotePlayBackChnoState(currentsession,channelstate,currentsession->channel))
		{
			//ʵʱ���͸���ͨ�������״̬
			if(currentsession->channelmask != channelstate || !currentsession->ifirst)
			{
				if(!currentsession->ifirst)
				{
					DEBUG_WARNING("---------channelstate:[%02X]------------------\n",channelstate);
					currentsession->ifirst		= 1;
					//channelstate					= currentsession->channel;
					//currentsession->channelmask	= 0x0;
				}
				DEBUG_WARNING("------channelmask:[%02X]-------channelstate:[%02X]-----sessionid:[%s]-getpid:[%d]-\n",currentsession->channelmask,channelstate,sessionid.c_str(),getpid());
				currentsession->channelmask 				= channelstate;
				Json::Value response;
				response[SESSION] 							= sessionid;
				response[MODULE] 								= MODULE_MEDIASTREAM;
				response[OPERATION] 							= MEDIA_REMOTECHANNELSTATUS;
				response[PARAMETER][MEDIASTREAM_PT]		= currentsession->pt;
				response[PARAMETER][MEDIASTREAM_SSRC] 		= currentsession->ssrc;
				response[PARAMETER][MEDIASTREAM_CHANNEL]	= currentsession->channel;
				response[PARAMETER][MEDIACHANNELMASK]		= channelstate;
				sender											=  packet.Serialize(0, currentsession->ssrc,response );
				CPSocketUtils::Send(taskHandle,sender.c_str(),sender.length(),0,500000,0);
			}
		}
		else
		{
			//DEBUG_WARNING("-----------------------------------\n");
			return -1;
		}
	}


	if(currentsession->stopflag)
	{
		return 0;
	}

	unsigned int	 flag				= 0;
	int result 						= 0;
	int	 freamtype						= currentsession->pt;
	int	 playbackchnindex				= 0;

	//ÿ�ξ�����ÿ��ͨ������ȡһ��
	int			count					= m_remoteplaybackchcount*3;
	//每次轮巡各个通道额数据
	while(count--)
	{
		if(currentsession->stopflag ||
			(currentsession->cmdflag == 2) ||
			(currentsession->cmdflag	== 3))
		{
			DEBUG_WARNING("-------stopflag:[%d]------------cmd[%d]-------\n",currentsession->stopflag,currentsession->cmdflag);
			return 0;
		}
		//由于数据所在的缓存是一个循环buf，有可能一帧数据被分割成2部分，为了保证数据的完整性，使用2个读指针标识数据的位子，通常first一直有效，send根据实际情况有时无效
		//һ֡����п����ڶ�ȡ��ʱ��ֳ���2���֣�datalenfirst��һ��������ݣ�����ʧ�ܣ�datalensecond��һ�������
		result		= GetRemotePlayBackData(currentsession,&databuffirst,datalenfirst,&databufsecond,datalensecond,formattype,freamtype,channel,flag,playbackchnindex);
//	DEBUG_WARNING("----result:[%d]--freamtype:[%d]--channel:[%d]-----\n",result,freamtype,channel);

		switch(result)
		{
		case 0://read success
			{
				//��ʽ��
				if(1 == formattype)
				{
					formattype			= 1;
					DEBUG_WARNING("--------------ͷ�ļ�--------------\n");
					return 0;
				}
				else
				{
					formattype	= currentsession->pt;
				}

#if 0
				if(channel == 0)
				{
					int	testFileFd		= -1;
					char	tmpTestFile[]	= "test.h264";
					testFileFd = open(tmpTestFile, O_RDWR|O_CREAT,777);

					if(testFileFd < 0)
					{
						DEBUG_WARNING("----------------------------\n");
					}
					else
					{
						chmod(tmpTestFile, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
					}
					lseek(testFileFd, 0, SEEK_END);
					write(testFileFd, databuffirst, datalenfirst);
					if(datalensecond > 0)
						write(testFileFd, databufsecond, datalensecond);
					close(testFileFd);
					testFileFd = -1;
				}
#endif

#if 0
//				FrameData_t	freamdata;
				//unsigned long long pts	= 0;
			//	DEBUG_WARNING("--streamtype:[%d]-chno:[%d]-channel:[%d]-\n",freamtype,currentsession->ssrc,channel);
				if(0 == freamtype && channel == 0)
				{

					if(datalenfirst > 44)
					{
//						freamdata.DataPack[0].pAddr[0] = (databuffirst + 44);
//						freamdata.DataPack[0].Len[0] = (datalenfirst - 44);
//						freamdata.DataPack[0].pAddr[1] = databufsecond;
//						freamdata.DataPack[0].Len[1] = datalensecond;
//						unsigned int	 check	= CalculateCheckSum(&freamdata,freamdata.DataPack[0].Len[0]  + freamdata.DataPack[0].Len[1]);
//
//						if(databuffirst[7] != check)
//						{
//							DEBUG_WARNING("--the-fream-check-fialed--type:[%d]-check:[%02X][%02X]--\n",freamtype,databuffirst[7],check);
//						}
//						DEBUG_WARNING("%d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d %d\n",
//								databuffirst[0],databuffirst[1],databuffirst[2],databuffirst[3],databuffirst[4],databuffirst[5],databuffirst[6],databuffirst[7],databuffirst[8],databuffirst[9],
//								databuffirst[10],databuffirst[11],databuffirst[12],databuffirst[13],databuffirst[14],databuffirst[15],databuffirst[16],databuffirst[17],databuffirst[18],databuffirst[19],
//								databuffirst[20],databuffirst[21],databuffirst[22],databuffirst[23],databuffirst[24],databuffirst[25],databuffirst[26],databuffirst[27],databuffirst[28],databuffirst[29],
//								databuffirst[30],databuffirst[31],databuffirst[32],databuffirst[33],databuffirst[34],databuffirst[35],databuffirst[36],databuffirst[37],databuffirst[38],databuffirst[39],
//								databuffirst[41],databuffirst[42],databuffirst[43],databuffirst[44]);
					//	memcpy((unsigned char*)&pts,&databuffirst[24],8);
						DEBUG_WARNING("chn:[%d]datatime:[20%d-%d-%d:%d:%d:%d]\n",channel,databuffirst[36],databuffirst[37],databuffirst[38],
								databuffirst[39],databuffirst[40],databuffirst[41]);
					}

				}
#endif
				//���ļ���read ,send flag to pc
				if(1 == flag)//next part start
				{
//					DEBUG_WARNING("-----------begin-----flag-:[%d]------------------------\n",flag);
					Json::Value Response;
					Response[MODULE] 										= MODULE_MEDIASTREAM;
					Response[OPERATION] 									= MEDIA_REMOTEPLAYBACKSTART;
					Response[SESSION] 									= sessionid;
					Response[RESPONSE][MEDIASTREAM_STREAMNAME]		= taskName;
					Response[RESPONSE][MEDIASTREAM_CSRC]				= currentsession->ssrc;
					Response[RESPONSE][MEDIASTREAM_PT] 				= currentsession->pt;
					Response[RESPONSE][ERRORCODE]  						= SUCCESS_OK;
					Response[RESPONSE][ERRORCAUSE] 						= EP_CN_SUCCESS_OK;
					Response[RESPONSE][MEDIA_REST]						= 2;
					Response[RESPONSE][MEDIASTREAM_STREAMTYPE]		= currentsession->streamtype;
					Response[RESPONSE][MEDIASTREAM_CHANNEL]			= currentsession->channel;
					Response[RESPONSE][MEDIACHANNELMASK]				= ( 0x01 << channel);
					sender.clear();
					sender = packet.Serialize(0, currentsession->ssrc,Response);
					if(CPSocketUtils::Send(taskHandle,sender.c_str(),sender.length(),0,5000000,0) < 0)
					{
						currentsession->stopflag	= 1;
						return -3;
					}
				}

				sender.clear();
				if(datalensecond > 0)
				{
					if(passiveMode)
					{
#ifndef NOT_SUPORT_ENOUGH_WIDTH
						sender = packet.Serialize(formattype, channel,datalenfirst + datalensecond);
						if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)databuffirst,sender.length(),datalenfirst,0,5000000,0) < 0)
						{
							return -3;//�����쳣������ʧ��
						}

						if(CPSocketUtils::Send(taskHandle,NULL,(char*)databufsecond,0,datalensecond,0,5000000,0) < 0)
						{
							return -3;//�����쳣������ʧ��
						}
#else
						int	   havesendlen		= 0;
						while(datalenfirst > havesendlen)
						{
							if(datalenfirst - havesendlen > MAX_SEND_LEN_PASSIVE)
							{
								//DEBUG_WARNING("----datalenfirst:[%d]--havesendlen:[%d]-----\n",datalenfirst,havesendlen);
								sender = packet.Serialize(formattype, channel,MAX_SEND_LEN_PASSIVE);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),MAX_SEND_LEN_PASSIVE,0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen	+= MAX_SEND_LEN_PASSIVE;
							}
							else
							{
								//DEBUG_WARNING("----datalenfirst:[%d]--havesendlen:[%d]-----\n",datalenfirst,havesendlen);
								sender = packet.Serialize(formattype, channel,datalenfirst - havesendlen);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),(datalenfirst - havesendlen),0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen	= datalenfirst;
							}
						}

						havesendlen			= 0;
						while(datalensecond > havesendlen)
						{
							if(datalensecond - havesendlen > MAX_SEND_LEN_PASSIVE)
							{
								sender = packet.Serialize(formattype, channel,MAX_SEND_LEN_PASSIVE);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databufsecond + havesendlen),sender.length(),MAX_SEND_LEN_PASSIVE,0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		+= MAX_SEND_LEN_PASSIVE;
							}
							else
							{
								sender = packet.Serialize(formattype, channel,datalensecond - havesendlen );
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databufsecond + havesendlen),sender.length(),(datalensecond - havesendlen),0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		= datalensecond;
							}
						}
#endif
					}
					else
					{
						int	   havesendlen		= 0;
						while(datalenfirst > havesendlen)
						{
							if(datalenfirst - havesendlen > MAX_SEND_LEN)
							{
								//DEBUG_WARNING("----datalenfirst:[%d]--havesendlen:[%d]-----\n",datalenfirst,havesendlen);
								sender = packet.Serialize(formattype, channel,MAX_SEND_LEN);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),MAX_SEND_LEN,0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen	+= MAX_SEND_LEN;
							}
							else
							{
								//DEBUG_WARNING("----datalenfirst:[%d]--havesendlen:[%d]-----\n",datalenfirst,havesendlen);
								sender = packet.Serialize(formattype, channel,datalenfirst - havesendlen);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),(datalenfirst - havesendlen),0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen	= datalenfirst;
							}
						}

						havesendlen			= 0;
						while(datalensecond > havesendlen)
						{
							if(datalensecond - havesendlen > MAX_SEND_LEN)
							{
								sender = packet.Serialize(formattype, channel,MAX_SEND_LEN);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databufsecond + havesendlen),sender.length(),MAX_SEND_LEN,0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		+= MAX_SEND_LEN;
							}
							else
							{
								sender = packet.Serialize(formattype, channel,datalensecond - havesendlen );
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databufsecond + havesendlen),sender.length(),(datalensecond - havesendlen),0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		= datalensecond;
							}
						}
					}
				}
				else
				{
					if(passiveMode)
					{
#ifndef	NOT_SUPORT_ENOUGH_WIDTH
						sender = packet.Serialize(formattype, channel,datalenfirst);

						if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)databuffirst,sender.length(),datalenfirst,0,5000000,0) < 0)
						{
							return -3;//�����쳣������ʧ��
						}
#else

						int	   havesendlen		= 0;
						while(datalenfirst > havesendlen)
						{
							if(datalenfirst - havesendlen > MAX_SEND_LEN_PASSIVE)
							{
								DEBUG_WARNING("----datalenfirst:[%d]--havesendlen:[%d]-----\n",datalenfirst,havesendlen);
								sender = packet.Serialize(formattype, channel,MAX_SEND_LEN_PASSIVE);

								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),MAX_SEND_LEN_PASSIVE,0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		+= MAX_SEND_LEN_PASSIVE;
							}
							else
							{
								sender = packet.Serialize(formattype, channel,datalenfirst - havesendlen);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),(datalenfirst - havesendlen),0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		= datalenfirst;
							}
						}
#endif
					}
					else
					{
						int	   havesendlen		= 0;
						while(datalenfirst > havesendlen)
						{
							if(datalenfirst - havesendlen > MAX_SEND_LEN)
							{
								sender = packet.Serialize(formattype, channel,MAX_SEND_LEN);

								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),MAX_SEND_LEN,0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		+= MAX_SEND_LEN;
							}
							else
							{
								sender = packet.Serialize(formattype, channel,datalenfirst - havesendlen);
								if(CPSocketUtils::Send(taskHandle,sender.c_str(),(char*)(databuffirst + havesendlen),sender.length(),(datalenfirst - havesendlen),0,5000000,0) < 0)
								{
									return -3;//�����쳣������ʧ��
								}
								havesendlen		= datalenfirst;
							}
						}
					}
				}

				if(2 == flag)//old part is read over
				{
//					DEBUG_WARNING("-----------end-----flag-:[%d]------------------------\n",flag);
					Json::Value Response;
					Response[MODULE] 										= MODULE_MEDIASTREAM;
					Response[OPERATION] 									= MEDIA_REMOTEPLAYBACKSTOP;
					Response[SESSION] 									= sessionid;
					Response[RESPONSE][MEDIASTREAM_STREAMNAME]		= taskName;
					Response[RESPONSE][MEDIASTREAM_CSRC]				= currentsession->ssrc;
					Response[RESPONSE][MEDIASTREAM_PT] 				= currentsession->pt;
					Response[RESPONSE][ERRORCODE]  						= SUCCESS_OK;
					Response[RESPONSE][ERRORCAUSE] 						= EP_CN_SUCCESS_OK;
					Response[RESPONSE][MEDIA_REST]						= 2;
					Response[RESPONSE][MEDIASTREAM_STREAMTYPE]		= currentsession->streamtype;
					Response[RESPONSE][MEDIASTREAM_CHANNEL]			= currentsession->channel;
					Response[RESPONSE][MEDIACHANNELMASK]				= ((~( 0x01 << channel)) & currentsession->channel);
					sender.clear();
					sender = packet.Serialize(0, currentsession->ssrc,Response);
					if(CPSocketUtils::Send(taskHandle,sender.c_str(),sender.length(),0,5000000,0) < 0)
					{
						currentsession->stopflag							= 1;
						return -3;
					}
				}

				ReleasePlayBackFream(currentsession,playbackchnindex);
			}
			break;
		case -1://fail
			{
			DEBUG_WARNING("-----------------is-----failed----------------\n");
				return -1;
			}
			break;
		case -2://no data
			{
				//DEBUG_WARNING("----------------no data----------------\n");
				CPSelectUtils::mSleep(10);
				ReleasePlayBackFream(currentsession,playbackchnindex);
				return 0;
			}
			break;
		case -3://over
			{

				DEBUG_WARNING("-----------------is-----over----------------\n");
				return -2;
			}
			break;
		case -4://reading
			{
				CPSelectUtils::mSleep(30);
				//DEBUG_WARNING("----------------reading--------------\n");
				return 0;
			}
			break;
		}
	}

	CPSelectUtils::mSleep(20);
	return 0;
}
   //./////////////////////////
void CStreamLayer::ControlRemotePlayback(Json::Value  request)
{

	int		paramiserror					= 0;
	Json::Value response;
	response[SESSION] 					= request[SESSION];
	response[MODULE] 						= MODULE_MEDIASTREAM;
	response[OPERATION] 					= request[OPERATION];
	response[RESPONSE][ERRORCODE] 		= SUCCESS_OK;
	response[RESPONSE][ERRORCAUSE] 		= EP_CN_SUCCESS_OK;
	response[RESPONSE][MEDIASTREAM_STREAMNAME]	= request[PARAMETER][MEDIASTREAM_STREAMNAME];

	if(request[PARAMETER].empty())
	{
		response[RESPONSE][ERRORCODE] 	= ANALYZEFAILED;
		response[RESPONSE][ERRORCAUSE]	= EP_CN_ANALYZEFAILED;
	//	//_sessionCallback(source,taskindex,response,0);
		DEBUG_WARNING("----------------------------------\n");
		return ;
	}

	int cmd						= 1;
	if(!request[PARAMETER][MEDIASTREAM_PALYBACKCMD].isNull())
	{
		cmd							= request[PARAMETER][MEDIASTREAM_PALYBACKCMD].asInt();
	}
	else
	{
		paramiserror				= 1;
	}

	int		pt						= (int)PAULOAD_REMOTE_PLAYBACK_TYPE_EM;
	if(!request[PARAMETER][MEDIASTREAM_PT].isNull())
	{
		pt							= request[PARAMETER][MEDIASTREAM_PT].asInt();
		if(pt != PAULOAD_REMOTE_PLAYBACK_TYPE_EM )
		{
			paramiserror			= 1;
		}
	}
//	else
//	{
//		paramiserror				= 1;
//	}

	int		ssrc					= 0;
	if(!request[PARAMETER][MEDIASTREAM_SSRC].isNull())
	{
		ssrc						= request[PARAMETER][MEDIASTREAM_SSRC].asInt();
	}

	if(ssrc < 0 || ssrc > 65535)
	{
		paramiserror			= 1;
	}

	std::string	starttime 		= "";
	if(3 == cmd )
	{
		if(!request[PARAMETER][MEDIASTREAM_STARTTIME].isNull())
		{
			starttime				= request[PARAMETER][MEDIASTREAM_STARTTIME].asString();
		}
		else
		{
			paramiserror			= 1;
		}
	}

	std::string	endtime		= "";
	if(!request[PARAMETER][MEDIASTREAM_ENDTIME].isNull())
	{
		endtime					= request[PARAMETER][MEDIASTREAM_ENDTIME].asString();
	}

	if(paramiserror)
	{
		response[RESPONSE][ERRORCODE] 		= ANALYZEFAILED;
		response[RESPONSE][ERRORCAUSE] 		= EP_CN_ANALYZEFAILED;
		//_sessionCallback(source,taskindex,response,0);
		DEBUG_WARNING("-----------------------------------------\n");
		return;
	}

	DateInfo_Tag	 		startremotetime;
	DateInfo_Tag			endremotetime;
	memset(&startremotetime,0,sizeof(DateInfo_Tag));
	memset(&endremotetime,0,sizeof(DateInfo_Tag));

	if(!starttime.empty())
	{
		TimeStringToDateInfo(starttime,startremotetime);
	}

	if(!endtime.empty())
	{
		TimeStringToDateInfo(endtime,endremotetime);
	}

	msglink_interface(taskHandle,cmd,0);

	unsigned int	pts						= 0;
	if(!request[PARAMETER][MEDIASTREAM_OFFSET].isNull())
	{
		pts									= request[PARAMETER][MEDIASTREAM_OFFSET].asInt();
	}

	taskSession_node *currentsession 	= NULL;
	int		ssid							= (( pt << 24)|( ssrc << 8));

	_taskmutex.ThreadLock();
	currentsession						= QuerySession(ssid);
	if(!currentsession)
	{
		response[RESPONSE][ERRORCODE]	= EXECUFAILED;
		response[RESPONSE][ERRORCAUSE] 	= EP_CN_EXECUFAILED;
	}
	else
	{
		currentsession->offset			= pts;
		switch(cmd)
		{
			case 0:// stop
				{
					DEBUG_WARNING("-----stop-----------------\n");
					currentsession->stopflag			= 1;
					currentsession->cmdflag			= 0;
					if(!passiveMode)
					{
						currentsession->stopflag		= 2;
					}

				}
				break;
			case 1:
				{
					DEBUG_WARNING("-----play-----------------\n");
					currentsession->cmdflag			= 1;
				}
				break;
			case 2:
				{
					DEBUG_WARNING("-------pause---------------\n");

					currentsession->cmdflag			= 2;
				}
				break;
			case 3:
				{
					DEBUG_WARNING("-----seek-----------------\n");
					currentsession->cmdflag			= 3;
					memcpy(&currentsession->startremotetime,&startremotetime,sizeof(DateInfo_Tag));
				}
				break;
			case 4:
				{

					DEBUG_WARNING("-----fast--forward---------------\n");
					currentsession->restflag			= 2;
				}
				break;
			case 5:
				{
					DEBUG_WARNING("-----normal-----------------\n");
					currentsession->restflag			= 3;
				}
				break;
			default:
				{
					response[RESPONSE][ERRORCODE] 	= UNKNOW;
					response[RESPONSE][ERRORCAUSE] 	= EP_CN_UNKNOW;
					currentsession->stopflag			= 1;
				}
				break;
		}

	}

	_taskmutex.ThreadUnlock();

	if(3 == cmd)
	{
		return ;
	}
}


